package demo.watermark;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

import java.time.Duration;

public class TimeStampDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime); // 1.12之后默认就是eventtime语义

        env.getConfig().setAutoWatermarkInterval(1000); // default 200ms

        SingleOutputStreamOperator<String> input =
                env.socketTextStream("localhost", 9999)
                        .assignTimestampsAndWatermarks(
                                WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(10)));

        SingleOutputStreamOperator<String> input1 =
                env.socketTextStream("localhost", 9999)
                        .assignTimestampsAndWatermarks(
                                new AssignerWithPunctuatedWatermarks<String>() {
                                    @Nullable
                                    @Override
                                    public Watermark checkAndGetNextWatermark(
                                            String lastElement, long extractedTimestamp) {
                                        return null;
                                    }

                                    @Override
                                    public long extractTimestamp(
                                            String element, long recordTimestamp) {
                                        return 0;
                                    }
                                });

        SingleOutputStreamOperator<String> input2 =
                env.socketTextStream("localhost", 9999)
                        .assignTimestampsAndWatermarks(
                                new AssignerWithPeriodicWatermarks<String>() {
                                    @Nullable
                                    @Override
                                    public Watermark getCurrentWatermark() {
                                        return null;
                                    }

                                    @Override
                                    public long extractTimestamp(
                                            String element, long recordTimestamp) {
                                        return 0;
                                    }
                                });
    }
}
